package com.sts.task;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.StrUtil;
import com.sts.common.utils.FunctionUtil;
import com.sts.sink.StsSink;
import com.sts.sink.model.BaseSinkConfig;
import com.sts.source.StsSource;
import com.sts.source.model.BaseSourceConfig;
import com.sts.transform.StsTransform;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import javax.swing.*;
import java.util.List;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

@Slf4j
@Getter
public class StsTask {

    @Setter
    private volatile boolean flag = true;

    private volatile int successCount = 0;

    private static final AtomicIntegerFieldUpdater<StsTask> fieldUpdater =
            AtomicIntegerFieldUpdater.newUpdater(StsTask.class, "successCount");

    public void updateTotalSuccessCount(StsTask stsTask, int count) {
        fieldUpdater.getAndAccumulate(stsTask, count, Integer::sum);
        if (logTextArea != null) {
            logTextArea.append(StrUtil.format("{} - sink success {}\n", DateUtil.now(), count));
        }
        log.info("{} - sink success {}", DateUtil.now(), count);
    }

    @Setter
    private JTextArea logTextArea;

    /**
     * source 配置
     */
    private final BaseSourceConfig sourceConfig;

    /**
     * sink 配置
     */
    private final BaseSinkConfig sinkConfig;

    /**
     * 写入类型, 默认是INSERT
     */
    private SinkTypeEnum sinkType = SinkTypeEnum.INSERT;

    private List<StsTransform<?>> stsTransformList;

    public StsTask(BaseSourceConfig sourceConfig, BaseSinkConfig sinkConfig) {
        this.sourceConfig = sourceConfig;
        this.sinkConfig = sinkConfig;
    }

    public StsTask(BaseSourceConfig sourceConfig, BaseSinkConfig sinkConfig, SinkTypeEnum sinkType) {
        this.sourceConfig = sourceConfig;
        this.sinkConfig = sinkConfig;
        this.sinkType = sinkType;
    }

    public StsTask(BaseSourceConfig sourceConfig, BaseSinkConfig sinkConfig, List<StsTransform<?>> stsTransformList) {
        this.sourceConfig = sourceConfig;
        this.sinkConfig = sinkConfig;
        this.stsTransformList = stsTransformList;
    }

    public StsTask(BaseSourceConfig sourceConfig, BaseSinkConfig sinkConfig, SinkTypeEnum sinkType, List<StsTransform<?>> stsTransformList) {
        this.sourceConfig = sourceConfig;
        this.sinkConfig = sinkConfig;
        this.sinkType = sinkType;
        this.stsTransformList = stsTransformList;
    }

    public void run() {
        // 统计耗时
        TimeInterval timer = DateUtil.timer();

        if (CollectionUtil.isNotEmpty(stsTransformList)) {
            FunctionUtil.init();
        }

        try {
            StsSource stsSource = sourceConfig.buildImpl();
            StsSink stsSink = sinkConfig.buildImpl(stsSource, this);

            switch (sinkType) {
                case INSERT -> stsSink.insert();
                case UPSERT -> stsSink.upsert();
            }
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            long interval = timer.interval();
            if (logTextArea != null) {
                logTextArea.append(StrUtil.format("{} - 总共成功导入: {} 条, 耗时(ms): {}\n", DateUtil.now(), successCount, interval));
            }
            log.info("{} - 总共成功导入: {} 条, 耗时(ms): {}", DateUtil.now(), successCount, interval);
        }
    }

}
